package org.apache.flink.runtime.minicluster;

import static org.apache.flink.configuration.ClusterOptions.PROCESS_WORKING_DIR_BASE;
import static org.apache.flink.util.Preconditions.checkNotNull;
import static org.apache.flink.util.Preconditions.checkState;

import java.io.IOException;
import java.net.InetAddress;
import java.net.InetSocketAddress;
import java.time.Duration;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
import java.util.List;
import java.util.Optional;
import java.util.UUID;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.Executor;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import java.util.function.Function;
import java.util.function.Supplier;
import java.util.stream.Collectors;

import javax.annotation.Nonnull;
import javax.annotation.Nullable;
import javax.annotation.concurrent.GuardedBy;

import org.apache.flink.annotation.Internal;
import org.apache.flink.annotation.VisibleForTesting;
import org.apache.flink.api.common.ConsumerID;
import org.apache.flink.api.common.ConsumingStatus;
import org.apache.flink.api.common.ResourceStatus;
import org.apache.flink.api.common.io.FileOutputFormat;
import org.apache.flink.api.common.time.Time;
import org.apache.flink.configuration.ClusterOptions;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.configuration.ConfigurationUtils;
import org.apache.flink.configuration.HighAvailabilityOptions;
import org.apache.flink.configuration.IllegalConfigurationException;
import org.apache.flink.runtime.allocation.ResourceAllocationStatistics;
import org.apache.flink.runtime.blob.BlobCacheService;
import org.apache.flink.runtime.blob.BlobServer;
import org.apache.flink.runtime.blob.BlobUtils;
import org.apache.flink.runtime.clusterframework.types.ResourceID;
import org.apache.flink.runtime.consumermanager.HighAvailabilityMode;
import org.apache.flink.runtime.entrypoint.ClusterEntrypointUtils;
import org.apache.flink.runtime.entrypoint.ClusterInformation;
import org.apache.flink.runtime.entrypoint.WorkingDirectory;
import org.apache.flink.runtime.entrypoint.master.DefaultMasterComponentFactory;
import org.apache.flink.runtime.entrypoint.master.MasterComponentHolder;
import org.apache.flink.runtime.entrypoint.master.MasterComponentOperations;
import org.apache.flink.runtime.externalresource.ExternalResourceInfoProvider;
import org.apache.flink.runtime.heartbeat.HeartbeatServices;
import org.apache.flink.runtime.highavailability.HighAvailabilityServices;
import org.apache.flink.runtime.highavailability.HighAvailabilityServicesFactory;
import org.apache.flink.runtime.highavailability.HighAvailabilityServicesUtils;
import org.apache.flink.runtime.highavailability.nonha.embedded.EmbeddedHaServices;
import org.apache.flink.runtime.highavailability.nonha.embedded.EmbeddedHaServicesWithLeadershipControl;
import org.apache.flink.runtime.highavailability.nonha.embedded.HaLeadershipControl;
import org.apache.flink.runtime.leaderretrieval.LeaderRetrievalService;
import org.apache.flink.runtime.messages.Acknowledge;
import org.apache.flink.runtime.metrics.MetricRegistry;
import org.apache.flink.runtime.metrics.MetricRegistryConfiguration;
import org.apache.flink.runtime.metrics.MetricRegistryImpl;
import org.apache.flink.runtime.metrics.ReporterSetup;
import org.apache.flink.runtime.metrics.groups.ProcessMetricGroup;
import org.apache.flink.runtime.metrics.util.MetricUtils;
import org.apache.flink.runtime.resourceconsumer.ResourceConsumerGateway;
import org.apache.flink.runtime.resourceconsumer.ConsumingMasterId;
import org.apache.flink.runtime.resourcemanager.ResourceManagerGateway;
import org.apache.flink.runtime.resourcemanager.ResourceManagerId;
import org.apache.flink.runtime.resourcemanager.StandaloneResourceManagerFactory;
import org.apache.flink.runtime.resourceprovider.ResourceProvider;
import org.apache.flink.runtime.resourceprovider.ResourceProviderRunner;
import org.apache.flink.runtime.rpc.FatalErrorHandler;
import org.apache.flink.runtime.rpc.RpcService;
import org.apache.flink.runtime.rpc.RpcSystem;
import org.apache.flink.runtime.rpc.RpcUtils;
import org.apache.flink.runtime.rpc.retriever.MetricQueryServiceRetriever;
import org.apache.flink.runtime.rpc.retriever.impl.RpcGatewayRetriever;
import org.apache.flink.runtime.rpc.retriever.impl.RpcMetricQueryServiceRetriever;
import org.apache.flink.runtime.security.token.DelegationTokenManager;
import org.apache.flink.util.AutoCloseableAsync;
import org.apache.flink.util.ExceptionUtils;
import org.apache.flink.util.ExecutorUtils;
import org.apache.flink.util.Reference;
import org.apache.flink.util.concurrent.ExecutorThreadFactory;
import org.apache.flink.util.concurrent.ExponentialBackoffRetryStrategy;
import org.apache.flink.util.concurrent.FutureUtils;
import org.apache.flink.util.function.BiFunctionWithException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/** MiniCluster to execute 弹性资源 locally. */
public class MiniCluster implements AutoCloseableAsync {

    private static final Logger LOG = LoggerFactory.getLogger(MiniCluster.class);

    /** The lock to guard startup / shutdown / manipulation methods. */
    private final Object lock = new Object();

    /** The configuration for this mini cluster. */
    private final MiniClusterConfiguration miniClusterConfiguration;

    private final Time rpcTimeout;

    @GuardedBy("lock")
    private final List<ResourceProvider> resourceProviders;

    private final TerminatingFatalErrorHandlerFactory
            providerManagerTerminatingFatalErrorHandlerFactory =
                    new TerminatingFatalErrorHandlerFactory();
    private final Supplier<Reference<RpcSystem>> rpcSystemSupplier;

    private CompletableFuture<Void> terminationFuture;

    @GuardedBy("lock")
    private MetricRegistryImpl metricRegistry;

    @GuardedBy("lock")
    private ProcessMetricGroup processMetricGroup;

    @GuardedBy("lock")
    private RpcService commonRpcService;

    @GuardedBy("lock")
    private ExecutorService ioExecutor;

    @GuardedBy("lock")
    private final Collection<RpcService> rpcServices;

    @GuardedBy("lock")
    private HighAvailabilityServicesFactory haServicesFactory;

    @GuardedBy("lock")
    private HighAvailabilityServices haServices;

    @GuardedBy("lock")
    private BlobServer blobServer;

    @GuardedBy("lock")
    private HeartbeatServices heartbeatServices;

    @GuardedBy("lock")
    private DelegationTokenManager delegationTokenManager;

    @GuardedBy("lock")
    private BlobCacheService blobCacheService;

    @GuardedBy("lock")
    private LeaderRetrievalService consumerLeaderRetriever;
    
    @GuardedBy("lock")
    private RpcGatewayRetriever<ConsumingMasterId, ResourceConsumerGateway> consumerGatewayRetriever;
    
    @GuardedBy("lock")
    private LeaderRetrievalService resourceManagerLeaderRetriever;

    @GuardedBy("lock")
    private Collection<MasterComponentOperations> masterComponents;

    @GuardedBy("lock")
    private RpcGatewayRetriever<ResourceManagerId, ResourceManagerGateway>
            		resourceManagerGatewayRetriever;

    @GuardedBy("lock")
    private RpcServiceFactory providerManagerRpcServiceFactory;

    @GuardedBy("lock")
    private WorkingDirectory workingDirectory;

    /** Flag marking the mini cluster as started/running. */
    private volatile boolean running;

    @GuardedBy("lock")
    private Reference<RpcSystem> rpcSystem;

    // ------------------------------------------------------------------------

    /**
     * Creates a new Flink mini cluster based on the given configuration.
     *
     * @param miniClusterConfiguration The configuration for the mini cluster
     */
    public MiniCluster(MiniClusterConfiguration miniClusterConfiguration) {
        this(
                miniClusterConfiguration,
                () -> Reference.owned(RpcSystem.load(miniClusterConfiguration.getConfiguration())));
    }

    public MiniCluster(
            MiniClusterConfiguration miniClusterConfiguration,
            Supplier<Reference<RpcSystem>> rpcSystemSupplier) {

        this.miniClusterConfiguration =
                checkNotNull(miniClusterConfiguration, "config may not be null");
        this.rpcServices =
                new ArrayList<>(
                        1 + 2
                      + miniClusterConfiguration.getNumTaskManagers()); // common + (JM + RM) + TMs
        this.masterComponents = new ArrayList<>(1);

        // There shouldn't be any lost messages between the MiniCluster and the Flink components
        // since they all run in the same process.
        this.rpcTimeout = RpcUtils.INF_TIMEOUT;
        this.terminationFuture = CompletableFuture.completedFuture(null);
        running = false;
        // 资源提供者
        this.resourceProviders = new ArrayList<>(miniClusterConfiguration.getNumTaskManagers());
        //
        this.rpcSystemSupplier = rpcSystemSupplier;
    }

    public ClusterInformation getClusterInformation() {
        synchronized (lock) {
            checkState(running, "MiniCluster is not yet running or has already been shut down.");
            return new ClusterInformation("localhost", blobServer.getPort());
        }
    }

    protected Executor getIOExecutor() {
        return ioExecutor;
    }

    // ------------------------------------------------------------------------
    //  life cycle
    // ------------------------------------------------------------------------

    /** Checks if the mini cluster was started and is running. */
    public boolean isRunning() {
        return running;
    }

    /**
     * Starts the mini cluster, based on the configured properties.
     *
     * @throws Exception This method passes on any exception that occurs during the startup of the
     *     mini cluster.
     */
    public void start() throws Exception {
        synchronized (lock) {
            checkState(!running, "MiniCluster is already running");

            LOG.info("Starting elastic-resource Mini Cluster");
            LOG.debug("Using configuration {}", miniClusterConfiguration);

            final Configuration configuration = miniClusterConfiguration.getConfiguration();
            // 是否使用单一RPC服务
            final boolean useSingleRpcService =
                    miniClusterConfiguration.getRpcServiceSharing() == RpcServiceSharing.SHARED;
            //
            try {
                workingDirectory =
                        WorkingDirectory.create(
                                ClusterEntrypointUtils.generateWorkingDirectoryFile(
                                        configuration,
                                        Optional.of(PROCESS_WORKING_DIR_BASE),
                                        "minicluster_" + ResourceID.generate()));

                initializeIOFormatClasses(configuration);

                rpcSystem = rpcSystemSupplier.get();

                // metircs
                LOG.info("Starting Metrics Registry");
                metricRegistry =
                        createMetricRegistry(
                                configuration,
                                rpcSystem.deref().getMaximumMessageSizeInBytes(configuration));

                // bring up all the RPC services
                LOG.info("Starting RPC Service(s)");
                final RpcServiceFactory masterComponentRpcServiceFactory;
                final RpcService metricQueryServiceRpcService;

                if (useSingleRpcService) {
                    // we always need the 'commonRpcService' for auxiliary calls
                    commonRpcService = createLocalRpcService(configuration, rpcSystem.deref());
                    final CommonRpcServiceFactory commonRpcServiceFactory =
                            new CommonRpcServiceFactory(commonRpcService);
                    providerManagerRpcServiceFactory = commonRpcServiceFactory;
                    masterComponentRpcServiceFactory = commonRpcServiceFactory;
                    metricQueryServiceRpcService =
                            MetricUtils.startLocalMetricsRpcService(
                                    configuration, rpcSystem.deref());
                } else {

                    // start a new service per component, possibly with custom bind addresses
                    final String consumerManagerExternalAddress =
                            miniClusterConfiguration.getJobManagerExternalAddress();
                    final String taskManagerExternalAddress =
                            miniClusterConfiguration.getTaskManagerExternalAddress();
                    final String jobManagerExternalPortRange =
                            miniClusterConfiguration.getJobManagerExternalPortRange();
                    final String taskManagerExternalPortRange =
                            miniClusterConfiguration.getTaskManagerExternalPortRange();
                    // 消费者绑定地址
                    final String jobManagerBindAddress =
                            miniClusterConfiguration.getJobManagerBindAddress();
                    final String taskManagerBindAddress =
                            miniClusterConfiguration.getTaskManagerBindAddress();

                    masterComponentRpcServiceFactory =
                            new DedicatedRpcServiceFactory(
                                    configuration,
                                    consumerManagerExternalAddress,
                                    jobManagerExternalPortRange,
                                    jobManagerBindAddress,
                                    rpcSystem.deref());
                    providerManagerRpcServiceFactory =
                            new DedicatedRpcServiceFactory(
                                    configuration,
                                    taskManagerExternalAddress,
                                    taskManagerExternalPortRange,
                                    taskManagerBindAddress,
                                    rpcSystem.deref());

                    // we always need the 'commonRpcService' for auxiliary calls
                    // bind to the JobManager address with port 0
                    commonRpcService =
                            createRemoteRpcService(
                                    configuration, jobManagerBindAddress, 0, rpcSystem.deref());
                    metricQueryServiceRpcService =
                            MetricUtils.startRemoteMetricsRpcService(
                                    configuration,
                                    commonRpcService.getAddress(),
                                    null,
                                    rpcSystem.deref());
                }

                metricRegistry.startQueryService(metricQueryServiceRpcService, null);

                processMetricGroup =
                        MetricUtils.instantiateProcessMetricGroup(
                                metricRegistry,
                                RpcUtils.getHostname(commonRpcService),
                                ConfigurationUtils.getSystemResourceMetricsProbingInterval(
                                        configuration));

                ioExecutor =
                        Executors.newFixedThreadPool(
                                ClusterEntrypointUtils.getPoolSize(configuration),
                                new ExecutorThreadFactory("mini-cluster-io"));

                haServicesFactory = createHighAvailabilityServicesFactory(configuration);

                haServices = createHighAvailabilityServices(configuration, ioExecutor);

                blobServer =
                        BlobUtils.createBlobServer(
                                configuration,
                                Reference.borrowed(workingDirectory.getBlobStorageDirectory()),
                                haServices.createBlobStore());
                blobServer.start();

                heartbeatServices = HeartbeatServices.fromConfiguration(configuration);

                blobCacheService =
                        BlobUtils.createBlobCacheService(
                                configuration,
                                Reference.borrowed(workingDirectory.getBlobStorageDirectory()),
                                haServices.createBlobStore(),
                                new InetSocketAddress(
                                        InetAddress.getLocalHost(), blobServer.getPort()));

                //
                startTaskManagers();

                MetricQueryServiceRetriever metricQueryServiceRetriever =
                        new RpcMetricQueryServiceRetriever(
                                metricRegistry.getMetricQueryServiceRpcService());

                setupMasterComponents(
                        configuration,
                        masterComponentRpcServiceFactory,
                        metricQueryServiceRetriever);

                resourceManagerLeaderRetriever = haServices.getResourceManagerLeaderRetriever();

         
                resourceManagerGatewayRetriever =
                        new RpcGatewayRetriever<>(
                                commonRpcService,
                                ResourceManagerGateway.class,
                                ResourceManagerId::fromUuid,
                                new ExponentialBackoffRetryStrategy(
                                        21, Duration.ofMillis(5L), Duration.ofMillis(20L)));
     

                resourceManagerLeaderRetriever.start(resourceManagerGatewayRetriever);

            } catch (Exception e) {
                // cleanup everything
                try {
                    close();
                } catch (Exception ee) {
                    e.addSuppressed(ee);
                }
                throw e;
            }

            // create a new termination future
            terminationFuture = new CompletableFuture<>();

            // now officially mark this as running
            running = true;

            LOG.info("Flink Mini Cluster started successfully");
        }
    }

    // *********master组件
    // 设置master组件
    @GuardedBy("lock")
    private void setupMasterComponents(
            Configuration configuration,
            RpcServiceFactory dispatcherResourceManagerComponentRpcServiceFactory,
            MetricQueryServiceRetriever metricQueryServiceRetriever)
            throws Exception {
        masterComponents.addAll(
                createMasterComponents(
                        configuration,
                        dispatcherResourceManagerComponentRpcServiceFactory,
                        blobServer,
                        heartbeatServices,
                        delegationTokenManager,
                        metricRegistry,
                        metricQueryServiceRetriever,
                        new ShutDownFatalErrorHandler()));

        FutureUtils.completeAll(
                        masterComponents.stream()
                                .map(MasterComponentOperations::getShutDownFuture)
                                .collect(Collectors.toList()))
                .whenComplete((ignored, exception) -> closeAsync());
    }

    @VisibleForTesting
    protected Collection<? extends MasterComponentOperations>
            createMasterComponents(
                    Configuration configuration,
                    RpcServiceFactory rpcServiceFactory,
                    BlobServer blobServer,
                    HeartbeatServices heartbeatServices,
                    DelegationTokenManager delegationTokenManager,
                    MetricRegistry metricRegistry,
                    MetricQueryServiceRetriever metricQueryServiceRetriever,
                    FatalErrorHandler fatalErrorHandler)
                    throws Exception {
        DefaultMasterComponentFactory masterComponentFactory =
                createMasterComponentFactory();
        
        final MasterComponentOperations masterComponent =
                masterComponentFactory.create(
                		new ConsumerID(),
                		"mini-consumer",
                        configuration,
                        ResourceID.generate(),
                        ioExecutor,
                        rpcServiceFactory.createRpcService(),
                        haServices,
                        blobServer,
                        heartbeatServices,
                        metricRegistry,
                        metricQueryServiceRetriever,
                        fatalErrorHandler);
        FutureUtils.assertNoException(
                masterComponent
                        .getShutDownFuture()
                        .thenCompose(
                                applicationStatus ->
                                        masterComponent.stopApplication(
                                                applicationStatus, null)));
        return Collections.singleton(masterComponent);
    }

    protected DefaultMasterComponentFactory
            createMasterComponentFactory() {
        return DefaultMasterComponentFactory.createMasterComponentFactory(
                null, StandaloneResourceManagerFactory.getInstance());
    }

    // ********高可用服务
    private HighAvailabilityServicesFactory createHighAvailabilityServicesFactory(
            Configuration configuration) {
        final HaServices customMiniClusterHaServicesMode = miniClusterConfiguration.getHaServices();
        if (customMiniClusterHaServicesMode == HaServices.WITH_LEADERSHIP_CONTROL) {
            // special feature of MiniClusters to allow the control of leadership
            // EmbeddedLeaderElection requires a single instance for leader election across multiple
            // JobManager instances on the same JVM (after FLINK-24038 was introduced); therefore,
            // SingletonHighAvailabilityServicesFactory is utilized here
            return new SingletonHighAvailabilityServicesFactory(
                    (config, embeddedLeaderElectionExecutor) ->
                            new EmbeddedHaServicesWithLeadershipControl(
                                    embeddedLeaderElectionExecutor));
        } else if (customMiniClusterHaServicesMode != HaServices.CONFIGURED) {
            throw new IllegalConfigurationException(
                    "Unknown HA Services Mode configured in MiniCluster configuration: "
                            + customMiniClusterHaServicesMode);
        }

        final HighAvailabilityMode highAvailabilityMode =
                HighAvailabilityMode.fromConfig(configuration);
        if (highAvailabilityMode == HighAvailabilityMode.NONE) {
            // basic EmbeddedLeaderElection requires a single instance for leader election across
            // multiple JobManager instances on the same JVM (after FLINK-24038 was introduced);
            // therefore, SingletonHighAvailabilityServicesFactory is utilized here
            return new SingletonHighAvailabilityServicesFactory(
                    (config, embeddedLeaderElectionExecutor) ->
                            new EmbeddedHaServices(embeddedLeaderElectionExecutor));
        } else {
            return new RegularHighAvailabilityServicesFactory();
        }
    }

    @VisibleForTesting
    protected HighAvailabilityServices createHighAvailabilityServices(
            Configuration configuration, Executor executor) throws Exception {
        return haServicesFactory.createHAServices(configuration, executor);
    }

    /**
     * Returns {@link HaLeadershipControl} if enabled.
     *
     * <p>{@link HaLeadershipControl} allows granting and revoking leadership of HA components, e.g.
     * JobManager. The method return {@link Optional#empty()} if the control is not enabled in
     * {@link MiniClusterConfiguration}.
     *
     * <p>Enabling this feature disables {@link HighAvailabilityOptions#HA_MODE} option.
     */
    public Optional<HaLeadershipControl> getHaLeadershipControl() {
        synchronized (lock) {
            return haServices instanceof HaLeadershipControl
                    ? Optional.of((HaLeadershipControl) haServices)
                    : Optional.empty();
        }
    }

    protected HighAvailabilityServices getHaServices() {
        return haServices;
    }

    /**
     * Shuts down the mini cluster, failing all currently executing jobs. The mini cluster can be
     * started again by calling the {@link #start()} method again.
     *
     * <p>This method shuts down all started services and components, even if an exception occurs in
     * the process of shutting down some component.
     *
     * @return Future which is completed once the MiniCluster has been completely shut down
     */
    @Override
    public CompletableFuture<Void> closeAsync() {
        synchronized (lock) {
            if (running) {
                LOG.info("Shutting down Flink Mini Cluster");
                try {
                    final long shutdownTimeoutMillis =
                            miniClusterConfiguration
                                    .getConfiguration()
                                    .getLong(ClusterOptions.CLUSTER_SERVICES_SHUTDOWN_TIMEOUT);
                    final int numComponents = 2 + miniClusterConfiguration.getNumTaskManagers();
                    final Collection<CompletableFuture<Void>> componentTerminationFutures =
                            new ArrayList<>(numComponents);

                    componentTerminationFutures.addAll(terminateTaskManagers());

                    componentTerminationFutures.add(shutDownResourceManagerComponents());

                    final FutureUtils.ConjunctFuture<Void> componentsTerminationFuture =
                            FutureUtils.completeAll(componentTerminationFutures);

                    final CompletableFuture<Void> metricSystemTerminationFuture =
                            FutureUtils.composeAfterwards(
                                    componentsTerminationFuture, this::closeMetricSystem);

                    final CompletableFuture<Void> rpcServicesTerminationFuture =
                            FutureUtils.composeAfterwards(
                                    metricSystemTerminationFuture, this::terminateRpcServices);

                    final CompletableFuture<Void> remainingServicesTerminationFuture =
                            FutureUtils.runAfterwards(
                                    rpcServicesTerminationFuture,
                                    this::terminateMiniClusterServices);

                    final CompletableFuture<Void> executorsTerminationFuture =
                            FutureUtils.composeAfterwards(
                                    remainingServicesTerminationFuture,
                                    () -> terminateExecutors(shutdownTimeoutMillis));

                    final CompletableFuture<Void> deleteDirectoriesFuture =
                            FutureUtils.runAfterwards(
                                    executorsTerminationFuture, this::deleteDirectories);

                    deleteDirectoriesFuture.whenComplete(
                            (Void ignored, Throwable throwable) -> {
                                if (throwable != null) {
                                    terminationFuture.completeExceptionally(
                                            ExceptionUtils.stripCompletionException(throwable));
                                } else {
                                    terminationFuture.complete(null);
                                }
                            });
                } finally {
                    running = false;
                }
            }

            return terminationFuture;
        }
    }

    private CompletableFuture<Void> closeMetricSystem() {
        synchronized (lock) {
            final ArrayList<CompletableFuture<Void>> terminationFutures = new ArrayList<>(2);

            if (processMetricGroup != null) {
                processMetricGroup.close();
                processMetricGroup = null;
            }

            // metrics shutdown
            if (metricRegistry != null) {
                terminationFutures.add(metricRegistry.closeAsync());
                metricRegistry = null;
            }

            return FutureUtils.completeAll(terminationFutures);
        }
    }

    // *******资源提供者
    @GuardedBy("lock")
    private void startTaskManagers() throws Exception {
        final int numTaskManagers = miniClusterConfiguration.getNumTaskManagers();

        LOG.info("Starting {} TaskManager(s)", numTaskManagers);

        for (int i = 0; i < numTaskManagers; i++) {
            startTaskManager();
        }
    }

    /**
     * Starts additional TaskManager process.
     *
     * <p>When the MiniCluster starts up, it always starts {@link
     * MiniClusterConfiguration#getNumTaskManagers} TaskManagers. All TaskManagers are indexed from
     * 0 to the number of TaskManagers, started so far, minus one. This method starts a TaskManager
     * with the next index which is the number of TaskManagers, started so far. The index always
     * increases with each new started TaskManager. The indices of terminated TaskManagers are not
     * reused after {@link #terminateTaskManager(int)}.
     */
    public void startTaskManager() throws Exception {
        synchronized (lock) {
            final Configuration configuration = miniClusterConfiguration.getConfiguration();

            final ResourceProvider taskExecutor =
            		ResourceProviderRunner.startResourceProvider(
                            configuration,
                            new ResourceID(UUID.randomUUID().toString()),
                            providerManagerRpcServiceFactory.createRpcService(),
                            haServices,
                            heartbeatServices,
                            metricRegistry,
                            blobCacheService,
                            useLocalCommunication(),
                            ExternalResourceInfoProvider.NO_EXTERNAL_RESOURCES,
                            workingDirectory.createSubWorkingDirectory("tm_" + resourceProviders.size()),
                            providerManagerTerminatingFatalErrorHandlerFactory.create(
                            		resourceProviders.size()),
                            null
                            );

            taskExecutor.start();
            resourceProviders.add(taskExecutor);
        }
    }

    @VisibleForTesting
    protected boolean useLocalCommunication() {
        return miniClusterConfiguration.getNumTaskManagers() == 1;
    }

    @VisibleForTesting
    public Configuration getConfiguration() {
        return miniClusterConfiguration.getConfiguration();
    }

    // HACK: temporary hack to make the changelog state backend tests work with forced
    // full snapshots. This option should be removed once changelog state backend supports forced
    // full snapshots
    @Internal private boolean overrideRestoreModeForChangelogStateBackend;

    @Internal
    public void overrideRestoreModeForChangelogStateBackend() {
        this.overrideRestoreModeForChangelogStateBackend = true;
    }

    @GuardedBy("lock")
    private Collection<? extends CompletableFuture<Void>> terminateTaskManagers() {
        final Collection<CompletableFuture<Void>> terminationFutures =
                new ArrayList<>(resourceProviders.size());
        for (int i = 0; i < resourceProviders.size(); i++) {
            terminationFutures.add(terminateTaskManager(i));
        }

        return terminationFutures;
    }

    /**
     * Terminates a TaskManager with the given index.
     *
     * <p>See {@link #startTaskManager()} to understand how TaskManagers are indexed. This method
     * terminates a TaskManager with a given index but it does not clear the index. The index stays
     * occupied for the lifetime of the MiniCluster and its TaskManager stays terminated. The index
     * is not reused if more TaskManagers are started with {@link #startTaskManager()}.
     *
     * @param index index of the TaskManager to terminate
     * @return {@link CompletableFuture} of the given TaskManager termination
     */
    public CompletableFuture<Void> terminateTaskManager(int index) {
        synchronized (lock) {
            final ResourceProvider taskExecutor = resourceProviders.get(index);
            return taskExecutor.closeAsync();
        }
    }

    // ------------------------------------------------------------------------
    //  消费和资源状态
    // ------------------------------------------------------------------------
    public CompletableFuture<ResourceStatus> getResourceStatus() {
        return runResourceManangerCommand(
        		resourceManagerGateway -> resourceManagerGateway.requestResourceOverview(rpcTimeout))
        		.thenApply(resourceOverview ->  ResourceStatus.from(resourceOverview.getNumberRegisteredSlots(), 
        				resourceOverview.getNumberRegisteredSlots() - resourceOverview.getNumberFreeSlots())
        			);
    }
    
    public CompletableFuture<ResourceAllocationStatistics> requestResourceAllocationStatistics() {
        return runResourceManangerCommand(
        		resourceManagerGateway ->
        			resourceManagerGateway.requestResourceOverview(rpcTimeout))
        		.thenApply(resourceOverview ->  ResourceAllocationStatistics.from(resourceOverview));
    }
    
    public CompletableFuture<Acknowledge> cancel() {
        return runConsumerCommand(
        		consumerGateway -> consumerGateway.cancel(rpcTimeout));
    }
    
    public CompletableFuture<ConsumingStatus> getConsumingStatus() {
        return runConsumerCommand(
        		resourceConsumerGateway -> resourceConsumerGateway.requestConsumingStatus(rpcTimeout)
        			);
    }

    // 消费和资源状态---命令
    private <T> CompletableFuture<T> runConsumerCommand(
            Function<ResourceConsumerGateway, CompletableFuture<T>> consumerCommand) {
        return getConsumerGatewayFuture()
                .thenApply(consumerCommand)
                .thenCompose(Function.identity());
    }

    @VisibleForTesting
    protected CompletableFuture<ResourceConsumerGateway> getConsumerGatewayFuture() {
        synchronized (lock) {
            checkState(running, "MiniCluster is not yet running or has already been shut down.");
            return consumerGatewayRetriever.getFuture();
        }
    }

    private <T> CompletableFuture<T> runResourceManangerCommand(
            Function<ResourceManagerGateway, CompletableFuture<T>> command) {
        return getResourceManagerGatewayFuture()
                .thenApply(command)
                .thenCompose(Function.identity());
    }
    
    private CompletableFuture<ResourceManagerGateway> getResourceManagerGatewayFuture() {
        synchronized (lock) {
            checkState(running, "MiniCluster is not yet running or has already been shut down.");
            return resourceManagerGatewayRetriever.getFuture();
        }
    }

    // ------------------------------------------------------------------------
    //  factories - can be overridden by subclasses to alter behavior
    // ------------------------------------------------------------------------

    /**
     * Factory method to create the metric registry for the mini cluster.
     *
     * @param config The configuration of the mini cluster
     * @param maximumMessageSizeInBytes the maximum message size
     */
    protected MetricRegistryImpl createMetricRegistry(
            Configuration config, long maximumMessageSizeInBytes) {
        return new MetricRegistryImpl(
                MetricRegistryConfiguration.fromConfiguration(config, maximumMessageSizeInBytes),
                ReporterSetup.fromConfiguration(
                        config, miniClusterConfiguration.getPluginManager()));
    }

    /**
     * Factory method to instantiate the remote RPC service.
     *
     * @param configuration Flink configuration.
     * @param bindAddress The address to bind the RPC service to.
     * @param bindPort The port range to bind the RPC service to.
     * @param rpcSystem
     * @return The instantiated RPC service
     */
    protected RpcService createRemoteRpcService(
            Configuration configuration, String bindAddress, int bindPort, RpcSystem rpcSystem)
            throws Exception {
        return rpcSystem
                .remoteServiceBuilder(configuration, bindAddress, String.valueOf(bindPort))
                .withBindAddress(bindAddress)
                .withBindPort(bindPort)
                .withExecutorConfiguration(RpcUtils.getTestForkJoinExecutorConfiguration())
                .createAndStart();
    }

    /**
     * Factory method to instantiate the remote RPC service.
     *
     * @param configuration Flink configuration.
     * @param externalAddress The external address to access the RPC service.
     * @param externalPortRange The external port range to access the RPC service.
     * @param bindAddress The address to bind the RPC service to.
     * @param rpcSystem
     * @return The instantiated RPC service
     */
    protected RpcService createRemoteRpcService(
            Configuration configuration,
            String externalAddress,
            String externalPortRange,
            String bindAddress,
            RpcSystem rpcSystem)
            throws Exception {
        return rpcSystem
                .remoteServiceBuilder(configuration, externalAddress, externalPortRange)
                .withBindAddress(bindAddress)
                .withExecutorConfiguration(RpcUtils.getTestForkJoinExecutorConfiguration())
                .createAndStart();
    }

    /**
     * Factory method to instantiate the local RPC service.
     *
     * @param configuration Flink configuration.
     * @param rpcSystem
     * @return The instantiated RPC service
     */
    protected RpcService createLocalRpcService(Configuration configuration, RpcSystem rpcSystem)
            throws Exception {
        return rpcSystem
                .localServiceBuilder(configuration)
                .withExecutorConfiguration(RpcUtils.getTestForkJoinExecutorConfiguration())
                .createAndStart();
    }

    // ------------------------------------------------------------------------
    //  Internal methods
    // ------------------------------------------------------------------------

    @GuardedBy("lock")
    private CompletableFuture<Void> shutDownResourceManagerComponents() {

        final Collection<CompletableFuture<Void>> terminationFutures =
                new ArrayList<>(masterComponents.size());

        for (MasterComponentOperations masterComponent : masterComponents) {
            terminationFutures.add(masterComponent.closeAsync());
        }

        final FutureUtils.ConjunctFuture<Void> dispatcherTerminationFuture =
                FutureUtils.completeAll(terminationFutures);

        return FutureUtils.runAfterwards(
                dispatcherTerminationFuture,
                () -> {
                    Exception exception = null;

                    synchronized (lock) {
                        if (resourceManagerLeaderRetriever != null) {
                            try {
                                resourceManagerLeaderRetriever.stop();
                            } catch (Exception e) {
                                exception = ExceptionUtils.firstOrSuppressed(e, exception);
                            }

                            resourceManagerLeaderRetriever = null;
                        }

                        if (consumerLeaderRetriever != null) {
                            try {
                                consumerLeaderRetriever.stop();
                            } catch (Exception e) {
                                exception = ExceptionUtils.firstOrSuppressed(e, exception);
                            }

                            consumerLeaderRetriever = null;
                        }

                    }

                    if (exception != null) {
                        throw exception;
                    }
                });
    }

    private void terminateMiniClusterServices() throws Exception {
        // collect the first exception, but continue and add all successive
        // exceptions as suppressed
        Exception exception = null;

        synchronized (lock) {
            if (blobCacheService != null) {
                try {
                    blobCacheService.close();
                } catch (Exception e) {
                    exception = ExceptionUtils.firstOrSuppressed(e, exception);
                }
                blobCacheService = null;
            }

            // shut down the blob server
            if (blobServer != null) {
                try {
                    blobServer.close();
                } catch (Exception e) {
                    exception = ExceptionUtils.firstOrSuppressed(e, exception);
                }
                blobServer = null;
            }

            // shut down high-availability services
            if (haServices != null) {
                try {
                    haServices.closeAndCleanupAllData();
                } catch (Exception e) {
                    exception = ExceptionUtils.firstOrSuppressed(e, exception);
                }
                haServices = null;
            }

            try {
                if (rpcSystem.isOwned()) {
                    rpcSystem.deref().close();
                }
            } catch (Exception e) {
                exception = ExceptionUtils.firstOrSuppressed(e, exception);
            }

            if (exception != null) {
                throw exception;
            }
        }
    }

    @Nonnull
    private CompletableFuture<Void> terminateRpcServices() {
        synchronized (lock) {
            final int numRpcServices = 1 + rpcServices.size();

            final Collection<CompletableFuture<?>> rpcTerminationFutures =
                    new ArrayList<>(numRpcServices);

            rpcTerminationFutures.add(commonRpcService.closeAsync());

            for (RpcService rpcService : rpcServices) {
                rpcTerminationFutures.add(rpcService.closeAsync());
            }

            commonRpcService = null;
            rpcServices.clear();

            return FutureUtils.completeAll(rpcTerminationFutures);
        }
    }

    private CompletableFuture<Void> terminateExecutors(long executorShutdownTimeoutMillis) {
        synchronized (lock) {
            if (ioExecutor != null) {
                return ExecutorUtils.nonBlockingShutdown(
                        executorShutdownTimeoutMillis, TimeUnit.MILLISECONDS, ioExecutor);
            } else {
                return CompletableFuture.completedFuture(null);
            }
        }
    }

    private void deleteDirectories() throws IOException {
        synchronized (lock) {
            if (workingDirectory != null) {
                workingDirectory.delete();
            }
        }
    }

    /** Internal factory for {@link RpcService}. */
    protected interface RpcServiceFactory {
        RpcService createRpcService() throws Exception;
    }

    /** Factory which returns always the common {@link RpcService}. */
    protected static class CommonRpcServiceFactory implements RpcServiceFactory {

        private final RpcService commonRpcService;

        CommonRpcServiceFactory(RpcService commonRpcService) {
            this.commonRpcService = commonRpcService;
        }

        @Override
        public RpcService createRpcService() {
            return commonRpcService;
        }
    }

    /** Factory which creates and registers new {@link RpcService}. */
    protected class DedicatedRpcServiceFactory implements RpcServiceFactory {

        private final Configuration configuration;
        private final String externalAddress;
        private final String externalPortRange;
        private final String bindAddress;
        private final RpcSystem rpcSystem;

        DedicatedRpcServiceFactory(
                Configuration configuration,
                String externalAddress,
                String externalPortRange,
                String bindAddress,
                RpcSystem rpcSystem) {
            this.configuration = configuration;
            this.externalAddress = externalAddress;
            this.externalPortRange = externalPortRange;
            this.bindAddress = bindAddress;
            this.rpcSystem = rpcSystem;
        }

        @Override
        public RpcService createRpcService() throws Exception {
            final RpcService rpcService =
                    MiniCluster.this.createRemoteRpcService(
                            configuration,
                            externalAddress,
                            externalPortRange,
                            bindAddress,
                            rpcSystem);

            synchronized (lock) {
                rpcServices.add(rpcService);
            }

            return rpcService;
        }
    }

    // ------------------------------------------------------------------------
    //  miscellaneous utilities
    // ------------------------------------------------------------------------

    private void initializeIOFormatClasses(Configuration configuration) {
        // TODO: That we still have to call something like this is a crime against humanity
        FileOutputFormat.initDefaultsFromConfiguration(configuration);
    }

    private class TerminatingFatalErrorHandler implements FatalErrorHandler {

        private final int index;

        private TerminatingFatalErrorHandler(int index) {
            this.index = index;
        }

        @Override
        public void onFatalError(Throwable exception) {
            // first check if we are still running
            if (running) {
                LOG.error("TaskManager #{} failed.", index, exception);

                synchronized (lock) {
                    resourceProviders.get(index).closeAsync();
                }
            }
        }
    }

    private class ShutDownFatalErrorHandler implements FatalErrorHandler {

        @Override
        public void onFatalError(Throwable exception) {
            LOG.warn("Error in MiniCluster. Shutting the MiniCluster down.", exception);
            closeAsync();
        }
    }

    private class TerminatingFatalErrorHandlerFactory {

        /**
         * Create a new {@link TerminatingFatalErrorHandler} for the {@link TaskExecutor} with the
         * given index.
         *
         * @param index into the {@link #taskManagers} collection to identify the correct {@link
         *     TaskExecutor}.
         * @return {@link TerminatingFatalErrorHandler} for the given index
         */
        @GuardedBy("lock")
        private TerminatingFatalErrorHandler create(int index) {
            return new TerminatingFatalErrorHandler(index);
        }
    }

    /** HA Services to use. */
    public enum HaServices {
        /** Uses the configured HA Services in {@link HighAvailabilityOptions#HA_MODE} option. */
        CONFIGURED,

        /**
         * Enables or disables {@link HaLeadershipControl} in {@link
         * MiniCluster#getHaLeadershipControl}.
         *
         * <p>{@link HaLeadershipControl} allows granting and revoking leadership of HA components.
         * Enabling this feature disables {@link HighAvailabilityOptions#HA_MODE} option.
         */
        WITH_LEADERSHIP_CONTROL
    }

    /**
     * SingletonHighAvailabilityServicesFactory is used for scenarios that are not truly high
     * available and rely on having a single HighAvailabilityServices object.
     */
    private static class SingletonHighAvailabilityServicesFactory
            implements HighAvailabilityServicesFactory {

        private final BiFunctionWithException<
                        Configuration, Executor, HighAvailabilityServices, Exception>
                creationCallback;

        @Nullable private HighAvailabilityServices haServices;

        public SingletonHighAvailabilityServicesFactory(
                BiFunctionWithException<
                                Configuration, Executor, HighAvailabilityServices, Exception>
                        creationCallback) {
            this.creationCallback = creationCallback;
        }

        @Override
        public HighAvailabilityServices createHAServices(
                Configuration configuration, Executor executor) throws Exception {
            if (haServices == null) {
                haServices = creationCallback.apply(configuration, executor);
            }

            return this.haServices;
        }
    }

    private class RegularHighAvailabilityServicesFactory
            implements HighAvailabilityServicesFactory {
        @Override
        public HighAvailabilityServices createHAServices(
                Configuration configuration, Executor executor) throws Exception {
            return HighAvailabilityServicesUtils.createAvailableOrEmbeddedServices(
                    configuration, executor, new ShutDownFatalErrorHandler());
        }
    }
}
